package javaVersion.learn.flinkWindows;

/**
 * @program: myFlink
 * @description: 翻滚窗口(无重叠数据)
 * @author: WincoMa
 * @create: 2020-07-20 15:52
 **/

import javaVersion.pojo.Pro1_Window_WordCountCart;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 按照时间来进行窗口划分,每次窗口的滑动举例等于窗口的长度,
 * 这样数据不会重复计算
 */
public class Pro1_rollWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> textStream = senv.socketTextStream("hadoop1", 9999);
        textStream.map(new MapFunction<String, Pro1_Window_WordCountCart>() {
            @Override
            public Pro1_Window_WordCountCart map(String value) throws Exception {
                String[] data = value.split(" ");
                return new Pro1_Window_WordCountCart(Integer.parseInt(data[0]),Integer.parseInt(data[1]));
            }
        })
                .keyBy(0)
                //在使用翻滚窗口时,只需要传递窗口大小即可.直接可调用tumbling window翻滚窗口重载方法
                .timeWindow(Time.seconds(5))
                .sum(1)
                .print();
        senv.execute();

    }
}
